package com.open.capacity.notice.event;


import java.util.Date;
import java.util.Iterator;
import java.util.List;

import org.apache.commons.collections.CollectionUtils;
import org.springframework.stereotype.Service;

import com.open.capacity.common.disruptor.DisruptorTemplate;
import com.open.capacity.common.utils.BatchUtil;
import com.open.capacity.notice.constant.Topic;
import com.open.capacity.notice.context.SendMessageJobContext;
import com.open.capacity.notice.dmo.SendMessageTask;
import com.open.capacity.notice.service.SendMessageTaskService;

import lombok.RequiredArgsConstructor;

/**
 *  三方消息发送事件
 */
@Service
@RequiredArgsConstructor
public class SendMessagePublisher {

	private final DisruptorTemplate disruptorTemplate;

	private final SendMessageTaskService sendMessageTaskService;

	public void publish(List<SendMessageTask> sendMessageTasks, SendMessageJobContext context) {
		if (CollectionUtils.isNotEmpty(sendMessageTasks)) {
			BatchUtil.batchHandle(sendMessageTasks.size(), 200, (offset, size) -> {
				int toIndex = Math.min(offset + size, sendMessageTasks.size());
				List<SendMessageTask> splitTasks = sendMessageTasks.subList(offset, toIndex);
				splitTasks.forEach(task ->  task.setSendTime(new Date()));
				sendMessageTaskService.saveOrUpdateBatch(splitTasks, 100);
				if (CollectionUtils.isNotEmpty(splitTasks)) {
					for (Iterator<SendMessageTask> it = splitTasks.iterator(); it.hasNext();) {
						SendMessageTask sendMessageTask = it.next();
						disruptorTemplate.publish(Topic.NAME,
								SendMessageEvent.builder().sendMessageTask(sendMessageTask).build(),
								SendMessageContext.builder().sendMessageJobContext(context).build());
					}
				}
			});
		}

	}
}
